use super::ContainerId;
use super::ContainerResult;
use super::HalfContainerInfo;
use super::container_position;
use super::chunk_container::{ChunkContainerWriter,ChunkContainerReader};
use super::retrieve_container_id;

use crate::constants::DefaultFingerprintType;
use crate::constants::{WRITE_BUFFER_LEN,CACHE_SIZE};
use crate::chunk::Chunk;
use crate::container::seek_err_from;
use crate::index::IndexStore;
use crate::index::sample::SAMPLING;
use crate::rechunk::ReChunk;
use crate::recipe::ChunkPointer;

use util::LruCache;

use std::fs::metadata;
//use std::fs::File;
use std::io::{Seek,SeekFrom};
use std::fs::OpenOptions;
use std::collections::HashMap;
use std::path::Path;
use std::collections::VecDeque;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::Instant;
use parking_lot::{Mutex,Condvar};

lazy_static! {
	pub static ref FIND_FLAG:Mutex<bool> = Mutex::new(true);
}

#[derive(Clone)]
pub enum RecentChunk {
	Chunk(Arc<Chunk>),
	ChunkPointer(ContainerId,u32),
}

pub struct ContainerStore {
	file:String,
	cond:Condvar,
	start_id:ContainerId,
	id:AtomicU64,
	find_in_recent_buffer:bool,
	cache:Mutex<LruCache<ChunkContainerReader>>,
	write_buf:Mutex<(bool,VecDeque<Arc<ChunkContainerWriter>>,HashMap<DefaultFingerprintType,RecentChunk>,Option<Arc<ChunkContainerWriter>>)>,
	wait_write_all:(Condvar,Mutex<bool>),
	write_time:Mutex<u128>,
}

//container init
impl ContainerStore {
	pub fn from_path<P: AsRef<Path>+?Sized>(path:&P,find_in_recent_buffer:bool) -> ContainerResult<Self> {
		let path = path.as_ref();
		let start_id = match metadata(path) {
			Ok(meta) => {
				retrieve_container_id(meta.len())
			}
			Err(_) => {
				0
			}
		};
		Ok(Self {
			file:path.to_str().unwrap().to_string(),
			cond:Condvar::new(),
			start_id:start_id,
			find_in_recent_buffer:find_in_recent_buffer,
			id:AtomicU64::new(start_id),
			cache:Mutex::new(LruCache::with_capacity(CACHE_SIZE,
				OpenOptions::new().read(true).write(true).create(true).open(path)?
			)),
			write_buf:Mutex::new((false,VecDeque::with_capacity(WRITE_BUFFER_LEN),HashMap::new(),None)),
			wait_write_all:(Condvar::new(),Mutex::new(false)),
			write_time:Mutex::new(0),
		})
	}


}

//container store write
impl ContainerStore {
	pub fn write_async(&self,c:Arc<ChunkContainerWriter>,index_store:&IndexStore) -> ContainerResult<()> {
		let mut write_buf = self.write_buf.lock();
		write_buf.1.push_back(c.clone());
		if self.find_in_recent_buffer {
			for chunk in c.items() {
				let fp = chunk.fp();
				write_buf.2.remove(fp);
			}
		}
		SAMPLING(index_store,&c)?;
		self.cond.notify_all();
		Ok(())
	}

	pub fn get_write_time(&self) -> u128 {
		*self.write_time.lock()
	}

	pub fn insert_into_recent_buffer(&self,fp:&DefaultFingerprintType,recent_chunk:RecentChunk) {
		let mut recent_buffer = self.write_buf.lock();
		recent_buffer.2.entry(*fp).or_insert(recent_chunk);
	}

	pub fn find_in_recent_buffer(&self,fp:&DefaultFingerprintType) -> Option<RecentChunk> {
		match self.write_buf.lock().2.get(fp) {
			Some(recent) => {
				if let RecentChunk::Chunk(chunk) = recent {
					chunk.set_unique();
				}
				Some(recent.clone()) 
			},
			None => { None }
		}
	}
	
	pub fn pop_buf(&self) -> Option<Arc<ChunkContainerWriter>> {
		let mut deque = self.write_buf.lock();
		if !deque.0&&deque.1.is_empty() {
			self.cond.wait(&mut deque);
		}
		let con = deque.1.pop_front();
		deque.3 = con.clone();
		con
	}
	
	pub fn set_no_more(&self) {
		self.write_buf.lock().0 = true;
		self.cond.notify_all();
		
	}

	pub	fn container_alloc(&self) -> Arc<ChunkContainerWriter> {
		Arc::new(ChunkContainerWriter::new(self.id.fetch_add(1, std::sync::atomic::Ordering::Relaxed)))
	}

	pub	fn container_alloc_with_info(&self,info:Option<HalfContainerInfo>) -> Arc<ChunkContainerWriter> {
		if let Some(info) = info {
			Arc::new(ChunkContainerWriter::half(self.id.fetch_add(1, std::sync::atomic::Ordering::Relaxed),info))
		} else {
			Arc::new(ChunkContainerWriter::new(self.id.fetch_add(1, std::sync::atomic::Ordering::Relaxed)))
		}
	}
	
	pub fn find_in_write_buf<F:Fn(&ChunkContainerWriter)->bool>(&self,f:F) -> Option<Arc<ChunkContainerWriter>> {
		let con_list = &self.write_buf.lock().1;
		for con in con_list.iter() {
			if f(con) {
				return Some(con.clone());
			}
		}
		None
	}

	pub fn find_chunk_in_write_buf(&self,id:ContainerId,fp:&DefaultFingerprintType) -> Option<Arc<Chunk>> {
		let write_buf_inner = self.write_buf.lock();
		let write_buf = &write_buf_inner.1;
		let cur_write = &write_buf_inner.3;
		for con in write_buf.iter().chain(cur_write.iter()) {
			if con.id() == id {
				if let Some(chunk) = con.find(fp) {
					return Some(chunk);
				} else {
					panic!("hanging chunk poniter");
				}
			} 
		}
		None
	}
	
	pub fn wait_write_all(&self) {
		let mut write_all = self.wait_write_all.1.lock();
		if !*write_all {
			self.wait_write_all.0.wait(&mut write_all);
		}
	}

	pub fn set_write_all(&self) {
		let mut write_all = self.wait_write_all.1.lock();
		*write_all = true;
		self.wait_write_all.0.notify_all();
	}
	
	pub fn find_id_in_write_buf(&self,id:ContainerId) -> Option<Arc<ChunkContainerWriter>> {
		let con_list = &self.write_buf.lock().1;
		if con_list.is_empty() {return None;}
		let first_id = con_list.front().unwrap().id();
		let last_id = con_list.back().unwrap().id();
		if id>=first_id && id<=last_id {
			Some(con_list[(id-first_id) as usize].clone()) 
		}else {
			None
		}
	}

	pub fn find_chunk_pointer_in_write_buf<P:ChunkPointer>(&self,fp:&DefaultFingerprintType) -> Option<Box<P>> {
		let write_buf = &self.write_buf.lock().1;
		for con in write_buf.iter() {
			if let Some(chunk_pointer) = con.find_chunk_pointer(fp) {
				return Some(chunk_pointer);
			}
		}
		None
	}	
	
	pub fn write(&self) -> ContainerResult<()> {
		let mut cur_id:ContainerId = self.start_id;
		let mut file = OpenOptions::new()
					.write(true)
					.open(&self.file)?;
		file.seek(SeekFrom::Start(container_position(self.start_id)))?;
		let mut cost_time:u128 = 0;
		loop {
			let mut con = self.pop_buf();
			if let Some(ref mut con) = con {
				assert_eq!(con.id(),cur_id);
				let now = Instant::now();
				file.seek(SeekFrom::Start(container_position(cur_id)))
					.map_err(seek_err_from)?;
				con.on_write(&mut file)?;
				cost_time += now.elapsed().as_micros();
			}else {
				break;
			}
			cur_id += 1;
		}
		*self.write_time.lock() = cost_time;
		Ok(())
	}
}

//container store read
impl ContainerStore {
	pub fn read_container(&self,id:ContainerId) -> ContainerResult<Arc<ChunkContainerReader>>{
		let mut cache = self.cache.lock();
		cache.hit_with_id(id)
	}

	pub fn find_in_read_cache<F:Fn(&ChunkContainerReader)->bool>(&self,f:F) -> Option<Arc<ChunkContainerReader>> {
		self.cache.lock().hit(f)
	}

	pub fn find_chunk_in_read_cache(&self,fp:&DefaultFingerprintType) -> Option<Arc<ReChunk>> {
		let read_cache = self.cache.lock();
		for con in read_cache.into_iter() {
			if let Some(chunk) = con.find(fp) {
				return Some(chunk);
			}
		}
		None
	}

	pub fn find_chunk_pointer_in_read_cache<P:ChunkPointer>(&self,fp:&DefaultFingerprintType) -> Option<Box<P>> {
		let read_cache = self.cache.lock();
		for con in read_cache.into_iter() {
			if let Some(chunk_pointer) = con.find_chunk_pointer(fp) {
				return Some(chunk_pointer);
			}
		}
		None
	}	
}

